package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.constant.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

/**
 * @author Felix
 * @date 2022/11/28
 * 查询维度的工具类
 */
public class DimUtil {
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName,Tuple2.of("id",id));
    }
    /**
     * 查询维度数据：使用旁路缓存进行优化
     * 旁路缓存
     *      思路：先从缓存中获取维度数据，如果缓存中存在要查找的维度，那么直接将缓存中的维度作为返回值返回(缓存命中)；
     *           如果缓存中不存在要查找的维度，那么发送请求到phoenix的维度表中去查询维度数据，并将查询的结果放到
     *           缓存中缓存起来，方便下次查询使用。
     *      选型：
     *          Redis
     *              性能较高，维护性好  √
     *          状态
     *              性能高，维护性差
     *      注意：
     *          为了避免冷数据常驻内存，需要设置缓存的失效时间
     *          如果业务数据库维度发生了变化，应该将缓存的维度清除掉
     *      Redis中设计
     *          类型: String
     *          key: dim:维度表表名:主键1_主键2
     *          expire：     1day
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接从Redis中查询维度的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");

        //拼接查询维度的SQL
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.DIM_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            redisKey.append(columnValue);
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimInfoJsonStr = null;
        JSONObject dimInfoJsonObj = null;

        //先从缓存中获取维度数据
        try {
            jedis = RedisUtil.getJedis();
            dimInfoJsonStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }

        if(StringUtils.isNotEmpty(dimInfoJsonStr)){
            //如果缓存中存在要查找的维度，那么直接将缓存中的维度作为返回值返回(缓存命中)
            dimInfoJsonObj = JSON.parseObject(dimInfoJsonStr);
        }else{
            //如果缓存中不存在要查找的维度，那么发送请求到phoenix的维度表中去查询维度数据
            //底层还是调用的phoenixUtil中queryList方法，只不过根据维度主键查询维度，查询的结果中只会有一条数据
            System.out.println("从phoenix表中查询维度的SQL:" + selectSql);
            List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
            if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
                dimInfoJsonObj = dimInfoJsonObjList.get(0);
                //并将查询的结果放到  缓存中缓存起来，方便下次查询使用
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimInfoJsonObj.toJSONString());
                }
            } else {
                System.out.println("~~从phoenix表中没有找到对应的维度数据~~");
            }
        }

        //释放资源
        if(jedis != null){
            System.out.println("~~~关闭Jedis客户端~~~");
            jedis.close();
        }

        return dimInfoJsonObj;
    }

    //从phoenix表中查询维度数据
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        //拼接查询维度的SQL
        StringBuilder selectSql = new StringBuilder("select * from " + GmallConfig.DIM_SCHEMA + "." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "'");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }

        //底层还是调用的phoenixUtil中queryList方法，只不过根据维度主键查询维度，查询的结果中只会有一条数据
        System.out.println("从phoenix表中查询维度的SQL:" + selectSql);
        List<JSONObject> dimInfoJsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject dimInfoJsonObj = null;
        if (dimInfoJsonObjList != null && dimInfoJsonObjList.size() > 0) {
            dimInfoJsonObj = dimInfoJsonObjList.get(0);
        } else {
            System.out.println("~~从phoenix表中没有找到对应的维度数据~~");
        }
        return dimInfoJsonObj;
    }

    public static void delCached(String tableName,String id){
        String redisKey = "dim:" + tableName.toLowerCase() + ":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            if(jedis != null){
                System.out.println("维度数据发生变化后，从Redis中删除维度并关闭Jedis客户端");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) {
        // JSONObject dimInfo = getDimInfoNoCache("dim_base_trademark", Tuple2.of("id","1"));
        // JSONObject dimInfo = getDimInfo("dim_base_trademark", Tuple2.of("id","1"));
        JSONObject dimInfo = getDimInfo("dim_base_trademark", "1");
        System.out.println(dimInfo);
    }


}
